package dyyx.stream;

import java.time.LocalDateTime;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCount {

	public static void main(String[] args) throws Exception {
		
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

	    System.out.println(env);	
	    System.out.println("parallelism="+env.getParallelism());	

        DataStream<String> text = env.fromElements(
            "hello java","hello tiger","hello tiger","cat cat cat","dog"
        );
                
        text.flatMap(new Splitter())
        // .keyBy(0)
        .keyBy(new MyKeySelector())
        .sum(1)
        .print()
        // .writeAsCsv("/tmp/streamWordCountResult",FileSystem.WriteMode.OVERWRITE, "\n", " ");
        ;
     
        env.execute();
        
        System.out.println("done,"+LocalDateTime.now());
	}
	
	private static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
  		private static final long serialVersionUID = 1L;

		@Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
        	String[]words = sentence.split("\\W+");
        	for (String word: words) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
	
	private static class MyKeySelector  implements KeySelector<Tuple2<String, Integer>,String>{	
		private static final long serialVersionUID = 1L;

		public String getKey(Tuple2<String, Integer> value) {
			return value.f0;
		}
	}

}
